package cn.witsky.rtp2ws.client.ws.netty;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import cn.witsky.rtp2ws.domain.Rtp;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketHandshakeException;
import io.netty.util.CharsetUtil;
import net.sf.json.JSONObject;

public class NettyWsClientHandler extends SimpleChannelInboundHandler<Object> {
	private static Logger logger = LoggerFactory.getLogger(NettyWsClientHandler.class);
	volatile WebSocketClientHandshaker handshaker;
	volatile ChannelPromise handshakeFuture;
	volatile Rtp rtp;
	volatile Thread currentThread=null;

	public NettyWsClientHandler() {

	}

	private String getClientId(Channel channnel) {
		if (rtp != null) {
			return rtp.getRtpId();
		} else {
			NettyClient client = NettyClient.reMoveNettyClient(channnel);
			if (client != null) {
				return client.getClientId();
			} else {
				return "null";
			}
		}
	}

	public void setHandshakeFuture(ChannelPromise handshakeFuture) {
		this.handshakeFuture = handshakeFuture;
	}

	public void setRtp(Rtp rtp) {
		this.rtp = rtp;
	}

//	 public NettyWsClientHandler(WebSocketClientHandshaker handshaker,Rtp rtp) {
//		this.handshaker=handshaker;
//		this.rtp=rtp;
//	}
	public void handlerAdded(ChannelHandlerContext ctx) {
		Channel ch = ctx.channel();
		logger.debug("task:{} 客户端{}  handshaker:{} handlerAdded", this.getClientId(ch), ch, handshaker);
		this.handshakeFuture = ctx.newPromise();
	}

	public WebSocketClientHandshaker getHandshaker() {
		return handshaker;
	}

	public void setHandshaker(WebSocketClientHandshaker handshaker) {
		this.handshaker = handshaker;
		if(this.currentThread!=null) {
			LockSupport.unpark(currentThread);
		}
	}

	public ChannelPromise getHandshakeFuture() {
		return handshakeFuture;
	}

	public void channelActive(ChannelHandlerContext ctx) {
		Channel ch = ctx.channel();
		logger.info("task:{} 客户端{}  handshaker:{}  channelActive", this.getClientId(ch), ch, handshaker);
		ctx.fireChannelActive();
	}

	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		Channel ch = ctx.channel();
		FullHttpResponse response;
	
		if(this.handshaker==null) {
			this.currentThread=Thread.currentThread();
			LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
			this.currentThread=null;
		}
		if (!this.handshaker.isHandshakeComplete()) {
			try {
				response = (FullHttpResponse) msg;
				// 握手协议返回，设置结束握手
				this.handshaker.finishHandshake(ch, response);
				// 设置成功
				this.handshakeFuture.setSuccess();
				logger.debug("task:{} 客户端{} WebSocket Client connected! response headers[sec-websocket-extensions]:{}",
						this.getClientId(ch), ch, response.headers());
//				logger.info("WebSocket Client connected! response headers[sec-websocket-extensions]:{}",response.headers());
			} catch (WebSocketHandshakeException var7) {
				FullHttpResponse res = (FullHttpResponse) msg;
				String errorMsg = String.format("WebSocket Client failed to connect,status:%s,reason:%s", res.status(),
						res.content().toString(CharsetUtil.UTF_8));
				this.handshakeFuture.setFailure(new Exception(errorMsg));
			}
		} else if (msg instanceof FullHttpResponse) {
			response = (FullHttpResponse) msg;
			// this.listener.onFail(response.status().code(),
			// response.content().toString(CharsetUtil.UTF_8));
			throw new IllegalStateException("Unexpected FullHttpResponse (getStatus=" + response.status() + ", content="
					+ response.content().toString(CharsetUtil.UTF_8) + ')');
		} else {
			WebSocketFrame frame = (WebSocketFrame) msg;
			if (frame instanceof TextWebSocketFrame) {
				TextWebSocketFrame textFrame = (TextWebSocketFrame) frame;
//				LogUtil.info(this.rtp, "NettyWsClientHandler[channelRead0:98]", "task:{} 客户端{}收到消息:{}.....",
//						this.getClientId(ch), ch, textFrame.text());
				NettyClient client = NettyClient.getNettyClient(ctx.channel());
				if (client != null) {
					JSONObject jsonMsg = null;
					if (client.isSubscribe()) {
						try {
							jsonMsg = JSONObject.fromObject(textFrame.text());
						} catch (Exception ex) {

						}
						if (jsonMsg != null) {
							if (client.getWsListener() != null) {
								client.getWsListener().onMessage(jsonMsg);
							}
						}

					}
				}

			} else if (frame instanceof BinaryWebSocketFrame) {
				logger.debug("BinaryWebSocketFrame");
			} else if (frame instanceof PongWebSocketFrame) {
				logger.debug("WebSocket Client received pong");
			} else if (frame instanceof CloseWebSocketFrame) {
				logger.debug("receive close frame");
			}

		}
	}

	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		Channel ch = ctx.channel();
		logger.info("task:{} 客户端{}离线了", this.getClientId(ch), ch);
		NettyClient client = NettyClient.reMoveNettyClient(ctx.channel());
		if (client != null) {
			if (client.isSubscribe()) {
				if (client.getWsListener() != null) {
					client.getWsListener().onClose();
				}
			}
		}
		ch.close();
		ctx.fireChannelInactive();
	}

	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		Channel ch = ctx.channel();
		logger.warn("task:{} 客户端{}出现异常:{}", this.getClientId(ch), ch, cause.getCause());
		NettyClient client = NettyClient.reMoveNettyClient(ctx.channel());
		if (client != null) {
			if (client.isSubscribe()) {
				if (client.getWsListener() != null) {
					client.getWsListener().onClose();
				}
			}
		}
		ch.close();
		ctx.fireExceptionCaught(cause);
	}
}